package com.lenovo.ukr


import java.util.Properties
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.tree.RandomForest
import org.apache.spark.mllib.tree.model.RandomForestModel
import org.apache.spark.sql.SparkSession
// $example off$

object read_mysql {

  case class Person(target: Int, band: Int, work_years: Int, country: Int, urgency: Int, complaint : Int, compliment : Int,SR : Int,QI : Int ,L0_readtimes : Int)

  def main(args: Array[String]): Unit = {
      val sparkSession = SparkSession.builder.master("local").appName("RFExample").enableHiveSupport().getOrCreate()
      val sc = sparkSession.sparkContext
      var test_user:Map[String,String] = Map()
      sparkSession.read.csv("用户知识等级标注数据.csv").rdd.map(item=>{item(0)+"#"+item(1)})
        .collect().toList.foreach(item =>{
        val level_num = level2num(item.toString.split("#")(1)).toString
        test_user += (item.toString.split("#")(0)-> level_num)
      })

    val url = "jdbc:mysql://60.205.171.171:3306/liup?user=root&password=123456";
    val prop = new Properties();
    val df = sparkSession.read.jdbc(url, "tb_upp_user_profile", prop);
    df.show()
    df.createTempView("mysql_tmp")

    import sparkSession.sqlContext.implicits._
    val all_test_user_df = sparkSession.sql("select user_name,band,working_years,country,urgency,complaint,praise,ticket_sr,ticket_qi,reading_times from mysql_tmp where user_name is not null AND user_name!='' AND LOWER(user_name)!= 'null'")
      .rdd.map(item =>{
      val itcode = item(0)+""
      val band = item(1)+""
      val work_years = item(2)+""
      val country = item(3)+""
      val urgency = item(4)+""
      val complaint = item(5)+""
      val compliment = item(6)+""
      val SR =  item(7)+""
      val QI = item(8)+""
      val L0_readtimes = item(9)+""
      List(itcode,convert_band(band),convert_work_years(work_years), convert_country(country), convert_urgency(urgency), convert_complaint(complaint),convert_compliment(compliment), convert_sr(SR),convert_qi(QI), convert_L0_readtimes(L0_readtimes))
    }).filter(item=> {test_user.contains(item(0).toString)})
    .map(item =>
    {Person(test_user(item(0).toString).toInt,item(1).toInt,item(2).toInt,item(3).toInt,item(4).toInt,
      item(5).toInt,item(6).toInt,item(7).toInt,item(8).toInt,item(9).toInt)
    }).toDF()

    all_test_user_df.show()
    //all_test_user_df.write.format("csv").save("data1.csv")

    val featInd = all_test_user_df.columns.diff(List("target")).map(all_test_user_df.columns.indexOf(_))
      // Get index of target
      val targetInd = all_test_user_df.columns.indexOf("target")
      val data  = all_test_user_df.rdd.map(r => LabeledPoint(
        // Get target value
        r.getInt(targetInd),
        // Map feature indices to values
        Vectors.dense(featInd.map(r.getInt(_).toDouble))
      ))
    val splits = data.randomSplit(Array(0.7, 0.3))
    val (trainingData, testData) = (splits(0), splits(1))
    // Train a RandomForest model.
    // Empty categoricalFeaturesInfo indicates all features are continuous.
    val numClasses = 4
    val categoricalFeaturesInfo = Map[Int, Int]()
    val numTrees = 3 // Use more in practice.
    val featureSubsetStrategy = "auto" // Let the algorithm choose.
    val impurity = "gini"
    val maxDepth = 4
    val maxBins = 32

    val model = RandomForest.trainClassifier(trainingData, numClasses, categoricalFeaturesInfo,
      numTrees, featureSubsetStrategy, impurity, maxDepth, maxBins)

    // Evaluate model on test instances and compute test error
    val labelAndPreds = testData.map { point =>
      val prediction = model.predict(point.features)
      (point.label, prediction)
    }
    val testErr = labelAndPreds.filter(r => r._1 != r._2).count.toDouble / testData.count()
    println("Test Error = " + testErr)
    println("Learned classification forest model:\n" + model.toDebugString)

    // Save and load model
    model.save(sc, "myRandomForestClassificationModel")
    val sameModel = RandomForestModel.load(sc, "myRandomForestClassificationModel")
    sparkSession.stop()
  }

def convert_band(band : String):String = {
  var all_band:Map[String,String] = Map()
  all_band+=("Slovakia"->1.toString)
  all_band+=("1"->"1")
  all_band+=("2"->"2")
  all_band+=("3"->"3")
  all_band+=("4"->"4")
  all_band+=("5"->"5")
  all_band+=("6"->"6")
  all_band+=("7"->"7")
  all_band+=("8"->"8")
  all_band+=("9"->"9")
  all_band+=("10"->"10")
  all_band+=("China Contractor (inactive)"->"11")
  all_band+=("CEO"->"12")
  all_band+=("10 P"->"13")
  all_band+=("Brazil Corporate Administration (inactive)"->"14")
  all_band+=("EVP"->"14")
  all_band+=("SVP"->"16")
  all_band+=("VP"->"17")
  all_band+=("Contract"->"18")
  all_band+=("ED"->"19")
  if (all_band.contains(band) ) return all_band(band).toString
  else if(band == null || ("").equals(band) || ("null").equals(band.toLowerCase)) return "0"
  else return "20"
}
  def convert_work_years(working_years:String):String={
  if(working_years == "5+") return "11"
  else if(working_years == null || ("").equals(working_years) || ("null").equals(working_years.toLowerCase)) return "0"
  else  return (working_years.toDouble*2).toInt.toString
}
  def convert_urgency(urgency:String):String={
  if(urgency == "Y") return "1"
  else  return "0"
}
  def convert_complaint(compliment:String):String={
  if(compliment == "Y") return "1"
  else  return "0"
}
  def convert_compliment(compliment:String):String={
  if(compliment == "Y") return "1"
  else  return "0"
}
  def convert_sr(sr:String):String={
  if(sr == "Y") return "1"
  else  return "0"
}
  def convert_qi(qi:String):String={
  if(qi == "Y") return "1"
  else  return "0"
}
  def convert_L0_readtimes(times:String):String={
  if(times == "High") return "3"
  else if(times == "Medium") return "2"
  else if(times == "Low") return "1"
  else  return "0"
}
  def convert_ticket_distribution(distribution:String):String={

  return "0"
}
  def convert_country(country : String):String = {
  var all_country:Map[String,String] = Map()
  all_country+=("Slovakia"->1.toString)
  all_country+=("UnitedKingdom"->2.toString)
  all_country+=("Denmark"->3.toString)
  all_country+=("CN"->4.toString)
  all_country+=("MY"->5.toString)
  all_country+=("Japan"->6.toString)
  all_country+=("US"->7.toString)
  all_country+=("Slovenia"->8.toString)
  all_country+=("Spain"->9.toString)
  all_country+=("Indonesia"->10.toString)
  all_country+=("Austria"->11.toString)
  all_country+=("JP"->12.toString)
  all_country+=("GB"->13.toString)
  all_country+=("Singapore"->14.toString)
  all_country+=("UnitedStatesofAmerica"->15.toString)
  all_country+=("India"->16.toString)
  all_country+=("Brazil"->17.toString)
  all_country+=("BR"->18.toString)
  all_country+=("China"->19.toString)
  all_country+=("Mexico"->20.toString)
  all_country+=("IN"->21.toString)
  all_country+=("Philippines"->22.toString)
  all_country+=("UnitedArabEmirates"->23.toString)
  all_country+=("Argentina"->24.toString)
  all_country+=("MX"->25.toString)
  all_country+=("Canada"->26.toString)
  all_country+=("Taiwan"->27.toString)
  all_country+=("France"->28.toString)
  all_country+=("SouthAfrica"->29.toString)
  all_country+=("SaudiArabia"->30.toString)
  all_country+=("TW"->31.toString)
  all_country+=("Colombia"->32.toString)
  all_country+=("Malaysia"->33.toString)
  all_country+=("RussianFederation"->34.toString)
  all_country+=("Peru"->35.toString)
  all_country+=("HK"->36.toString)
  all_country+=("Germany"->37.toString)
  all_country+=("Switzerland"->38.toString)
  all_country+=("KZ"->39.toString)
  all_country+=("Australia"->40.toString)
  all_country+=("Israel"->41.toString)
  all_country+=("DE"->42.toString)
  all_country+=("Romania"->43.toString)
  all_country+=("Norway"->44.toString)
  all_country+=("Chile"->45.toString)
  all_country+=("CO"->46.toString)
  all_country+=("PL"->47.toString)
  all_country+=("Sweden"->48.toString)
  all_country+=("RU"->49.toString)
  all_country+=("UA"->50.toString)
  all_country+=("Finland"->51.toString)
  all_country+=("Italy"->52.toString)
  all_country+=("RO"->53.toString)
  all_country+=("Ireland"->54.toString)
  all_country+=("Ukraine"->55.toString)
  all_country+=("Belgium"->56.toString)
  all_country+=("Netherlands"->57.toString)
  all_country+=("SK"->58.toString)
  all_country+=("Portugal"->59.toString)
  all_country+=("PH"->60.toString)
  all_country+=("AR"->61.toString)
  all_country+=("FR"->62.toString)
  all_country+=("HongKong"->63.toString)
  all_country+=("Thailand"->64.toString)
  all_country+=("Morocco"->65.toString)
  all_country+=("CL"->66.toString)
  all_country+=("Venezuela"->67.toString)
  all_country+=("HU"->68.toString)
  all_country+=("MA"->69.toString)
  all_country+=("TH"->70.toString)
  all_country+=("SA"->71.toString)
  all_country+=("Turkey"->72.toString)
  all_country+=("CzechRepublic"->73.toString)
  all_country+=("ZA"->74.toString)
  all_country+=("ID"->75.toString)
  all_country+=("GR"->76.toString)
  all_country+=("CA"->77.toString)
  all_country+=("Egypt"->78.toString)
  all_country+=("SV"->79.toString)
  all_country+=("AE"->80.toString)
  all_country+=("SG"->81.toString)
  all_country+=("TR"->82.toString)
  all_country+=("AU"->83.toString)
  all_country+=("Kazakhstan"->84.toString)
  all_country+=("Poland"->85.toString)
  all_country+=("SE"->86.toString)
  all_country+=("IE"->87.toString)
  all_country+=("ES"->88.toString)
  all_country+=("SI"->89.toString)
  all_country+=("KR"->90.toString)
  all_country+=("CZ"->91.toString)
  all_country+=("CH"->92.toString)
  all_country+=("EG"->93.toString)
  all_country+=("Vietnam"->94.toString)
  all_country+=("Hungary"->95.toString)
  all_country+=("IL"->96.toString)
  all_country+=("PE"->97.toString)
  all_country+=("Pakistan"->98.toString)
  all_country+=("DK"->99.toString)
  all_country+=("NL"->100.toString)
  all_country+=("LT"->101.toString)
  all_country+=("LU"->102.toString)
  all_country+=("NG"->103.toString)
  all_country+=("BE"->104.toString)
  all_country+=("NO"->105.toString)
  all_country+=("Serbia"->106.toString)
  all_country+=("Croatia"->107.toString)
  all_country+=("Nigeria"->108.toString)
  all_country+=("AT"->109.toString)
  all_country+=("VE"->110.toString)
  all_country+=("IT"->111.toString)
  all_country+=("NewZealand"->112.toString)
  all_country+=("Greece"->113.toString)
  all_country+=("Korea,Republicof"->114.toString)
  all_country+=("Kenya"->115.toString)
  all_country+=("PT"->116.toString)
  all_country+=("HR"->117.toString)
  all_country+=("VN"->118.toString)
  all_country+=("BG"->119.toString)
  all_country+=("Bulgaria"->120.toString)
  all_country+=("KE"->121.toString)
  all_country+=("NZ"->122.toString)
  all_country+=("YU"->123.toString)
  all_country+=("PK"->124.toString)
  all_country+=("FI"->125.toString)
  all_country+=("GE"->126.toString)
  all_country+=("GT"->127.toString)
  all_country+=("AF"->128.toString)
  all_country+=("TN"->129.toString)
  if (all_country.contains(country) ) return all_country(country).toString
  else if(country == null || ("").equals(country) || ("null").equals(country.toLowerCase)) return "0"
  else return "130"
}
  def level2num (lev : String):Int = {
  if(lev == "初级") return 1
  else if (lev == "中级" )return 2
  else if (lev == "高级" )return 3
  else return 0
}


}
// scalastyle:on println

